package main

import (
	"context"
	"crypto/hmac"
	"crypto/sha1"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	"pack.ag/amqp"
)

// 参数说明，请参见AMQP客户端接入说明文档。
const consumerGroupId = "xAfn5d1vXaCkuY2pWdSN000100"
const clientId = "xufei-saas"

// iotInstanceId：实例ID。
const iotInstanceId = "iot-060a51yw"

// 接入域名，请参见AMQP客户端接入说明文档。
const host = "1240188612256500.iot-amqp.cn-shanghai.aliyuncs.com"

func main() {
	//工程代码泄露可能会导致 AccessKey 泄露，并威胁账号下所有资源的安全性。以下代码示例使用环境变量获取 AccessKey 的方式进行调用，仅供参考
	accessKey := "LTAI5tCx9QXqBSenuri5Rzmg"
	accessSecret := "XMOQ75sBtsOToryf9AfcMsiIGUj5th"
	address := "amqps://" + host + ":5671"
	timestamp := time.Now().Nanosecond() / 1000000
	//userName组装方法，请参见AMQP客户端接入说明文档。
	userName := fmt.Sprintf("%s|authMode=aksign,signMethod=Hmacsha1,consumerGroupId=%s,authId=%s,iotInstanceId=%s,timestamp=%d|",
		clientId, consumerGroupId, accessKey, iotInstanceId, timestamp)
	stringToSign := fmt.Sprintf("authId=%s&timestamp=%d", accessKey, timestamp)
	hmacKey := hmac.New(sha1.New, []byte(accessSecret))
	hmacKey.Write([]byte(stringToSign))
	//计算签名，password组装方法，请参见AMQP客户端接入说明文档。
	password := base64.StdEncoding.EncodeToString(hmacKey.Sum(nil))

	amqpManager := &AmqpManager{
		address:  address,
		userName: userName,
		password: password,
	}

	//如果需要做接受消息通信或者取消操作，从Background衍生context。
	ctx := context.Background()

	amqpManager.startReceiveMessage(ctx)
}

type DeviceReceived struct {
	DeviceName interface{} `json:"deviceName"`
	Status     interface{} `json:"status"`
	ChangeTime interface{} `json:"time"`
}

// 业务函数。用户自定义实现，该函数被异步执行，请考虑系统资源消耗情况。
func (am *AmqpManager) processMessage(ctx context.Context, message *amqp.Message) {
	// fmt.Println("data received:", string(message.GetData()), " properties:", message.ApplicationProperties)
	//defer am.receiver.Close(ctx)
	dataByte := message.GetData()
	data := string(dataByte)
	var device DeviceReceived
	// fmt.Println(data)
	if len(data) > 15 {
		// deviceName := gjson.Get(data, "deviceName")
		// status := gjson.Get(data, "status")
		// changeTime := gjson.Get(data, "time")
		err := json.Unmarshal(message.GetData(), &device)
		if err != nil {
			fmt.Println(err)
		}

	}
	if len(data) == 10 {
		in_no := data[5:6]
		status := data[7:8]
		topic, ok := message.ApplicationProperties["topic"].(string)
		if !ok {
			return
		}
		generateTime, ok := message.ApplicationProperties["generateTime"].(int64)
		if !ok {
			return
		}

		messageId, ok := message.ApplicationProperties["messageId"].(int64)
		if !ok {
			return
		}

		tm := time.Unix(generateTime/1000, 0)
		timeTemplate := "2006-01-02 15:04:05"
		timeStr := tm.Format(timeTemplate)

		topicArr := strings.Split(topic, "/")
		deviceName := topicArr[2]

		messageIdStr := fmt.Sprintf("Message:%d", messageId)
		// already, err := redisdb.Exists(ctx, messageIdStr).Result()
		// fmt.Println("12312", already)
		// if err != nil {
		// 	return
		// }
		// if already == 0 {
		fmt.Println(messageIdStr, "时间：", timeStr, " 设备编号:", deviceName, " 端口:", in_no, " 状态:", status)
		var url string

		// switch platform {
		// case "xufei":
		// 	url = "http://admin.xufeizizhu.com"
		// 	break
		// case "xufei-test":
		// 	url = "http://xufei11.400539.com"
		// 	break
		// case "chuse":
		// 	url = "http://cszzxc.com"
		// 	break
		// case "chuse-test":
		// 	url = "http://xufei22.400539.com"
		// 	break
		// case "smart":
		// 	url = "https://billiard.xufeizizhu.com"
		// 	break
		// case "smart-test":
		// 	url = "http://taiqiu.xufeizizhu.com/"
		// 	break
		// default:
		// 	fmt.Printf(">> platform >> %s | %s", deviceName, platform)
		// 	break
		// }

		// if len(url) != 0 {
		inNo, err := strconv.Atoi(in_no)
		if err != nil {
			fmt.Println(">> in no >>", err)
			// 可以根据实际情况进行更合适的错误处理，比如返回错误给调用者

		}
		statusInt, err := strconv.Atoi(status)
		if err != nil {
			fmt.Println(">> status >>", err)
			// 同样可以根据实际情况进行更合适的错误处理

		}
		// 后续使用inNo和statusInt变量进行操作
		fmt.Println(">> Get请求 >> ", deviceName)
		fmt.Printf(">> inNo %d >> status %d >>\n", inNo, statusInt)
		go func() {
			switch [2]int{inNo, statusInt} {
			case [2]int{1, 1}:
				url = "https://admin.cszzxc.com/api/newcar/button_switch_new?device_id=" + deviceName
				fmt.Println(url)
				response, err1 := Get(url)

				if err1 != nil {
					return
				}
				if response.StatusCode == 200 {
					// fmt.Println(messageIdStr, platform, ":入场请求成功")
					// redisdb.Set(ctx, messageIdStr, time.Now().String()+":go", time.Minute*60)
				} else {
					// fmt.Println(messageIdStr, platform, ":入场请求失败：", response)
				}

			case [2]int{2, 1}:
				url = "https://admin.cszzxc.com/api/newcar/out_ground_new?device_id=" + deviceName
				// fmt.Println(url)
				response, err1 := Get(url)
				if err1 != nil {
					return
				}
				if response.StatusCode == 200 {
					// fmt.Println(messageIdStr, platform, ":开门请求成功")
					// redisdb.Set(ctx, messageIdStr, time.Now().String()+":go", time.Minute*60)
				} else {
					// fmt.Println(messageIdStr, platform, ":开门请求失败：", response)
				}
			case [2]int{3, 1}:
				url = "https://admin.cszzxc.com/api/newcar/towel?device_id=" + deviceName
				// fmt.Println(url)
				response, err1 := Get(url)
				if err1 != nil {
					return
				}
				if response.StatusCode == 200 {
					// fmt.Println(messageIdStr, platform, ":开门请求成功")
					// redisdb.Set(ctx, messageIdStr, time.Now().String()+":go", time.Minute*60)
				} else {
					// fmt.Println(messageIdStr, platform, ":开门请求失败：", response)
				}
			default:

			}

		}()
	}
}

type AmqpManager struct {
	address  string
	userName string
	password string
	client   *amqp.Client
	session  *amqp.Session
	receiver *amqp.Receiver
}

func (am *AmqpManager) startReceiveMessage(ctx context.Context) {

	childCtx, _ := context.WithCancel(ctx)
	err := am.generateReceiverWithRetry(childCtx)
	if nil != err {
		return
	}
	defer func() {
		am.receiver.Close(childCtx)
		am.session.Close(childCtx)
		am.client.Close()
	}()

	for {
		//阻塞接受消息，如果ctx是background则不会被打断。
		message, err := am.receiver.Receive(ctx)

		if nil == err {
			go am.processMessage(ctx, message)
			message.Accept()
		} else {
			fmt.Println("amqp receive data error:", err)

			//如果是主动取消，则退出程序。
			select {
			case <-childCtx.Done():
				return
			default:
			}

			//非主动取消，则重新建立连接。
			err := am.generateReceiverWithRetry(childCtx)
			if nil != err {
				return
			}
		}
	}
}

func (am *AmqpManager) generateReceiverWithRetry(ctx context.Context) error {
	//退避重连，从10ms依次x2，直到20s。
	duration := 10 * time.Millisecond
	maxDuration := 20000 * time.Millisecond
	times := 1

	//异常情况，退避重连。
	for {
		select {
		case <-ctx.Done():
			return amqp.ErrConnClosed
		default:
		}

		err := am.generateReceiver()
		if nil != err {
			time.Sleep(duration)
			if duration < maxDuration {
				duration *= 2
			}
			fmt.Println("amqp connect retry,times:", times, ",duration:", duration)
			times++
		} else {
			fmt.Println("amqp connect init success")
			return nil
		}
	}
}

// 由于包不可见，无法判断Connection和Session状态，重启连接获取。
func (am *AmqpManager) generateReceiver() error {

	if am.session != nil {
		receiver, err := am.session.NewReceiver(
			amqp.LinkSourceAddress("/queue-name"),
			amqp.LinkCredit(20),
		)
		//如果断网等行为发生，Connection会关闭导致Session建立失败，未关闭连接则建立成功。
		if err == nil {
			am.receiver = receiver
			return nil
		}
	}

	//清理上一个连接。
	if am.client != nil {
		am.client.Close()
	}

	client, err := amqp.Dial(am.address, amqp.ConnSASLPlain(am.userName, am.password))
	if err != nil {
		return err
	}
	am.client = client

	session, err := client.NewSession()
	if err != nil {
		return err
	}
	am.session = session

	receiver, err := am.session.NewReceiver(
		amqp.LinkSourceAddress("/queue-name"),
		amqp.LinkCredit(20),
	)
	if err != nil {
		return err
	}
	am.receiver = receiver

	return nil
}
func Get(url string) (res *http.Response, err error) {
	// http client
	client := new(http.Client)
	client.Timeout = time.Second * 5
	res, err = client.Get(url)
	defer res.Body.Close()
	if err != nil {
		return nil, err
	}
	return
}
